package com.rpc.core.register.impl;

import com.rpc.core.holder.rag.RegInfoEnhancer;
import com.rpc.core.holder.rag.RegInfoHolder;
import com.rpc.core.holder.rag.RegInfo;
import com.rpc.core.register.RpcRegister;
import com.rpc.core.util.GsonUtil;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;

public class ZkRegister implements RpcRegister, ApplicationContextAware {

    private AtomicBoolean isStart=new AtomicBoolean(false);
    private Logger logger = LoggerFactory.getLogger(ZkRegister.class);

    private ApplicationContext context;

    private ZkClient zkClient;

    private String address;

    private String rootPath = "/rpc";

    private int connectionTimeout = 30000;

    private Map<String, RegInfoHolder> holderMap = new ConcurrentHashMap<>();

    public String getRootPath() {
        return rootPath;
    }

    public void setRootPath(String rootPath) {
        this.rootPath = rootPath;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public int getConnectionTimeout() {
        return connectionTimeout;
    }

    public void setConnectionTimeout(int connectionTimeout) {
        this.connectionTimeout = connectionTimeout;
    }

    @Override
    public void registered(RegInfo info) {

        //注册道
        String appName = info.getAppName();


        String path = this.rootPath + "/" + appName;
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path);
        }
        String uuid = UUID.randomUUID().toString();
        info.setUuid(uuid);
        String data = GsonUtil.GsonString(info);
        zkClient.createEphemeral(path + "/" + uuid, data);

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (!isStart.getAndSet(true)) {
            zkClient = new ZkClient(this.getAddress(), this.getConnectionTimeout());
            logger.info("zk connection init....");
            if (!zkClient.exists(this.getRootPath())) {
                zkClient.createPersistent(this.getRootPath());
            }
        }

    }

    @Override
    public void close() {
        if (this.zkClient != null&& isStart.getAndSet(false)) {
            this.zkClient.close();
        }
    }

    //监听变更事件
    @Override
    public RegInfoHolder listen(String appName) {

        String tag = this.getRootPath() + "/" + appName;
        if (holderMap.containsKey(tag)) {
            return holderMap.get(tag);
        }

         holderMap.computeIfAbsent(tag, key -> {
            RegInfoHolder regInfoHolder = new RegInfoHolder();

             List<String> children = zkClient.getChildren(tag);

             CopyOnWriteArrayList<RegInfoEnhancer> infos = new CopyOnWriteArrayList<>();
             for (String child : children) {
                 String regInfoString = (String) zkClient.readData(key + "/" + child);
                 RegInfo regInfo = GsonUtil.GsonToBean(regInfoString, RegInfo.class);
                 RegInfoEnhancer enhancer = new RegInfoEnhancer(regInfo, context);
                 infos.add(enhancer);
             }
             regInfoHolder.setEnInfos(infos);
             return regInfoHolder;
        });

        this.zkClient.subscribeChildChanges(tag, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {

                RegInfoHolder holder = holderMap.get(s);
                synchronized (holder) {

                    logger.info("{} change", s);

                    List<RegInfoEnhancer> infos = holder.getEnInfos();
                    CopyOnWriteArrayList<RegInfoEnhancer> regInfos = new CopyOnWriteArrayList<>();
                    ArrayList<RegInfoEnhancer> removeList = new ArrayList<>();
                    infos.stream().forEach(info -> {
                        String uuid = info.getInfo().getUuid();
                        //还存活
                        if (list.contains(uuid)) {
                            regInfos.add(info);
                        }else {
                            removeList.add(info);
                        }
                        list.remove(uuid);
                    });

                    //list 中还剩新增的
                    for (String uuid : list) {
                        String regInfoString = (String) zkClient.readData(s + "/" + uuid);
                        RegInfo regInfo = GsonUtil.GsonToBean(regInfoString, RegInfo.class);
                        RegInfoEnhancer enhancer = new RegInfoEnhancer(regInfo, context);
                        regInfos.add(enhancer);
                    }
                    holder.setEnInfos(regInfos);

                    for (RegInfoEnhancer enhancer : removeList) {
                        enhancer.close();
                        logger.info("{} try close",enhancer.getInfo().getAddress());
                    }

                }
            }
        });

        return holderMap.get(tag);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context=applicationContext;
    }
}
